package cn.gwm.flink.streaming.function.process;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import cn.gwm.flink.streaming.strategy.multidimension.MultiDimensionTimePredict;
import cn.gwm.flink.streaming.strategy.vehiclepredict.RedisInfo;
import cn.gwm.utils.HBaseDataSource;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;

/**
 * @author : GW00280745
 * @date : 2022/12/1 14:03
 * @description :
 */
public class MultiModePredictProcessFunction extends KeyedProcessFunction<Object, JSONObject, JSONObject> {
    @Override
    public void processElement(JSONObject line,
                               KeyedProcessFunction<Object, JSONObject, JSONObject>.Context ctx,
                               Collector<JSONObject> out) throws Exception {
        try {
            //以充电开始点作为基础点 之后每隔单位soc进行一次预测
            String vinVal = line.getString(ChargePredictFields.ins().srcFieldVin);
            Long tidVal = line.getLong(ChargePredictFields.ins().srcFieldTid);
            Long startChargeTime = line.getLong(ChargePredictFields.ins().srcFieldStartChargeTime);
            Float bms_soc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
            Float bms_curr = line.getFloat(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
            String chargeGroupName = vinVal.concat(ChargePredictFields.ins().char3).concat(startChargeTime.toString());
            String stsValue = tidVal.toString().concat(ChargePredictFields.ins().char3).concat(bms_soc.toString());
            //判断状态中是否有数据，如果没有那么就是第一个间隔点
            if (!latestSOCIntervalPointState.contains(chargeGroupName)){
                //满足条件进行计算
//                String predictTime = timePredict.time(line, jedisCluster);
                String predictTime = null;
                line.put(ChargePredictFields.ins().stdFieldPredictTimeV2,predictTime);
                //更新状态
                latestSOCIntervalPointState.put(chargeGroupName,stsValue);
                latestEstimateState.put(chargeGroupName,predictTime);
                //当前充电的第一次预测
                line.put(ChargePredictFields.ins().stdFieldEstiFirstPoint,ChargePredictFields.ins().oneStr);
                //第一个点也是间隔点，所以标识为1 标识这个预测值是计算出的不是复制的值
                line.put(ChargePredictFields.ins().stdFieldEstiPoint,ChargePredictFields.ins().oneStr);
                return;
            }
            //状态中有数据，那么进行间隔点条件比较
            BigDecimal socValue = new BigDecimal(bms_soc);
            Boolean isIntervalPoint = intervalPointCheck(chargeGroupName, socValue,latestSOCIntervalPointState);
            if (!isIntervalPoint){
                //不是间隔点标识为0
                line.put(ChargePredictFields.ins().stdFieldEstiPoint,ChargePredictFields.ins().zeroStr);
                //不是间隔点需要复制之前的预测值
                line.put(ChargePredictFields.ins().stdFieldPredictTime,latestEstimateState.get(chargeGroupName));
                return;
            }
            //符合间隔点条件 进行间隔点预算
//            String predictTime = timePredict.time(line);
            String predictTime = "";
            line.put(ChargePredictFields.ins().stdFieldPredictTimeV2,predictTime);
            //更新状态
            latestSOCIntervalPointState.put(chargeGroupName,stsValue);
            latestEstimateState.put(chargeGroupName,predictTime);
        }catch (Exception e){
            e.printStackTrace();
            line.put(ChargePredictFields.ins().stdFieldPredictTimeV2,e.getMessage());
        }
        out.collect(line);
    }
    /**
     * @description:  间隔点条件判断
     * @author:  GW00280745
     * @date: 2022/5/19 11:00
     * @param: chargeGroupName
     * @param: socValue
     * @return: java.lang.Boolean
     **/
    private Boolean intervalPointCheck(String chargeGroupName, BigDecimal socValue,
                                       MapState<String, String> latestSOCIntervalPointState) throws Exception{
        String latestStsValue = latestSOCIntervalPointState.get(chargeGroupName);
        String[] sp = latestStsValue.split(ChargePredictFields.ins().char3);
        String latestTidStr = sp[0];
        BigDecimal latestSocValue = new BigDecimal(sp[1]);
        String socStepStr = "5.0";
        BigDecimal socStep = new BigDecimal(socStepStr);
        BigDecimal socLim = latestSocValue.add(socStep);
        if (socValue.compareTo(socLim) >= 0){
            return true;
        }else {
            return false;
        }
    }
    private boolean invalidVal(JSONObject line){
        Float bms_soc = line.getFloat(ChargePredictFields.ins().srcFieldBmsSoc);
        Float bms_curr = line.getFloat(ChargePredictFields.ins().srcFieldBmsBatterCurrRm);
        Float acambtemp= line.getFloat("acambtemp");
        if (Math.abs(bms_curr)<3 || bms_soc == -999999f || acambtemp == -999999f){
            //等价标记一个更新点 用于解决电流波动带来的影响
            line.put("predict_time_v1", "null#curr_lt3#soc_temp_999");
            line.put(ChargePredictFields.ins().stdFieldEstiPoint,ChargePredictFields.ins().zeroStr);
            return true;
        }
        return false;
    }
    private MapState<String, String> latestSOCIntervalPointState;
    private MapState<String, String> latestEstimateState;
    private MultiDimensionTimePredict timePredict;
    private JedisCluster jedisCluster;
    @Override
    public void open(Configuration parameters) throws Exception {
        latestSOCIntervalPointState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("latestSOCIntervalPointState", String.class, String.class));
        latestEstimateState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("latestEstimateState", String.class, String.class));
        timePredict = new MultiDimensionTimePredict();
        HBaseDataSource.ins().getDataSource().getConnection();
        jedisCluster = RedisInfo.getInstance().jedisPoolAli();
    }
}
